package sink4mysql;
import java.sql.*;
import java.text.SimpleDateFormat;
import java.util.Date;

import net.sf.json.JSONObject;
import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Preconditions;

public class sink4mysql  extends AbstractSink implements Configurable{

    private static final Logger logger = LoggerFactory.getLogger(sink4mysql.class);
    private Connection connect;
    private PreparedStatement pstmt;
    private String sql_command;
    private String url;
    private String user;
    private String password;


    // 在整个sink结束时执行一遍
    @Override
    public synchronized void stop() {
        // TODO Auto-generated method stub
        super.stop();
    }

    // 在整个sink开始时执行一遍，用来初始化数据库连接
    @Override
    public synchronized void start() {
        // TODO Auto-generated method stub
        super.start();
        try {
            connect = DriverManager.getConnection(url,user,password);
            logger.debug("msg: Connect mysql ok");
            // 连接URL为 jdbc:mysql//服务器地址/数据库名 ，后面的2个参数分别是登陆用户名和密码
            pstmt = connect.prepareStatement(sql_command);
        } catch (SQLException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

    }

    // 不断循环调用，处理消息Event（本例就是插入数据库）
    public Status process() throws EventDeliveryException {

        // TODO Auto-generated method stub
        //事务，获取event什么的都是模板。仿照别的sink写就OK
        Channel ch = getChannel();
        Transaction txn = ch.getTransaction();
        Event event = null;
        txn.begin();
//        logger.debug("222222222222222222222222222222222");
        while (true) {
            event = ch.take();
            if (event != null) {
                break;
            }
        }
//        logger.debug("7777777777777777777777777777777777777777777777");
        try {
            String rawbody = new String(event.getBody());
//            JSONObject jsonObject = JSONObject.fromObject(rawbody);
            logger.debug("rawbody:"+rawbody);
            String body = rawbody;
            logger.debug("spiltbody:"+body);
            if (body.split(",").length>0) {
//                String sql = "insert into " + tableName + "(" + columnName + ") values(" + body + ")";
                //logger.error("sql:"+sql);
                pstmt.setString(1,body.split(",")[0]);
                pstmt.setString(2,body.split(",")[1]);
                pstmt.setString(3,body.split(",")[2]);

                pstmt.executeUpdate();
                logger.debug("msg: Insert data ok!");
                txn.commit();
                return Status.READY;
            } else {
                txn.rollback();
                return null;
            }
        } catch (Throwable th) {
            txn.rollback();
//            logger.debug("--------------------------------------------------");
            if (th instanceof Error) {
                throw (Error) th;
            } else {
                throw new EventDeliveryException(th);
            }
        } finally {
            txn.close();
        }

    }
    //从配置文件中读取各种属性，并进行一些非空验证
    public void configure(Context context) {

        url = context.getString("url");
        Preconditions.checkNotNull(url, "url must be set!!");
//        按照指定语句插入数据
        sql_command = context.getString("sqlcommand");
        Preconditions.checkNotNull(sql_command, "sqlcommand must be set!!");
        password = context.getString("password");
        Preconditions.checkNotNull(password, "password must be set!!");
        user = context.getString("user");
        Preconditions.checkNotNull(user, "user must be set!!");
        // Preconditions.checkNotNull(password, "password must be set!!");
        logger.debug("Config:"+url+","+sql_command);
    }
    public String getTime(){
        SimpleDateFormat sdf = new SimpleDateFormat();// 格式化时间
        sdf.applyPattern("yyyy-MM-dd HH:mm:ss");// a为am/pm的标记
        java.util.Date date = new Date();// 获取当前时间
        return sdf.format(date);
    }
}